package com.atguigu.gulimall.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.atguigu.gulimall.common.to.mq.SecKillOrderTo;
import com.atguigu.gulimall.common.utils.R;
import com.atguigu.gulimall.common.vo.MemberVo;
import com.atguigu.gulimall.feign.CouponFeignService;
import com.atguigu.gulimall.feign.ProductFeignService;
import com.atguigu.gulimall.inteceptor.KillInteceptor;
import com.atguigu.gulimall.service.SeckillService;
import com.atguigu.gulimall.to.SeckillSkuRedisTo;
import com.atguigu.gulimall.vo.SkuInfoVo;
import com.atguigu.gulimall.vo.SkuKillSessionVo;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import jdk.nashorn.internal.parser.Token;
import net.sf.jsqlparser.statement.select.KSQLJoinWindow;
import org.apache.commons.lang.StringUtils;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Service
public class SeckillServiceImpl implements SeckillService {

    public static final String SESSION_CACHE_PREFIX = "seckill:sessions:";
    public static final String SKUKILL_CACHE_PREFIX = "seckill:skus:";
    public static final String SKU_STOCK_SEMAPHORE = "seckill:stock:";
    public static final String BUY_ONE_KEY = "user_buy_one:";

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private ProductFeignService productFeignService;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private CouponFeignService couponFeignService;


    @Override
    public void uploadSeckillSkuLatest3Days() {
        R r = couponFeignService.productUp();
        Integer code = (Integer) r.get("code");
        if (code == 0) {
            List<SkuKillSessionVo> vos = (List<SkuKillSessionVo>) r.getData("list", new TypeReference<List<SkuKillSessionVo>>() {
            });
            // 1. 将秒杀场次存入redis中
            saveSessionInfos(vos);
            // 2. 缓存活动的关联商品信息
            saveSessionSkuInfos(vos);
        }
    }


    // 保存每个场次的所有秒杀商品
    private void saveSessionSkuInfos(List<SkuKillSessionVo> vos) {
        vos.stream().forEach(item -> {
            // 准备 hash操作
            BoundHashOperations<String, Object, Object> boundHashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
            item.getRelationVos().stream().forEach(vo -> {

                String token = UUID.randomUUID().toString().replace("-", "");

                //设置为幂等的

                if (!boundHashOps.hasKey((vo.getPromotionSessionId() + "_" + vo.getSkuId().toString()))) {
                    // 缓存商品
                    SeckillSkuRedisTo redisTo = new SeckillSkuRedisTo();
                    // 1. sku的基本信息
                    R info = productFeignService.info(vo.getSkuId());
                    if ((Integer) info.get("code") == 0) {
                        SkuInfoVo skuInfo = (SkuInfoVo) info.getData("skuInfo", new TypeReference<SkuInfoVo>() {
                        });
                        redisTo.setSkuInfoVo(skuInfo);
                    }

                    // 2.sku的秒杀信息
                    BeanUtils.copyProperties(vo, redisTo);

                    // 3.保存商品的起始与结束时间
                    redisTo.setStartTime(item.getStartTime());
                    redisTo.setEndTime(item.getEndTime());

                    // 4.保存一个随机ID，避免攻击
                    redisTo.setUuid(token);

                    String jsonString = JSON.toJSONString(redisTo);
                    boundHashOps.put(vo.getPromotionSessionId() + "_" + vo.getSkuId().toString(), jsonString);
                    // 5.使用redission的分布式信号量来锁定库存
                    RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + token);
                    // 将商品秒杀数量设置为信号量
                    semaphore.trySetPermits(vo.getSeckillCount().intValue());
                }

            });
        });
    }

    // 保存场次
    private void saveSessionInfos(List<SkuKillSessionVo> vos) {
        vos.stream().forEach(item -> {
            long startTime = item.getStartTime().getTime();
            long endTime = item.getEndTime().getTime();
            // key为前缀+场次起始时间+“_”+场次结束时间
            String key = SESSION_CACHE_PREFIX + startTime + "_" + endTime;

            // 判断该key是否存在，不存在则保存到redis
            Boolean hasKey = redisTemplate.hasKey(key);
            if (!hasKey) {

                List<String> collect = item.getRelationVos().stream().map(vo -> {
                    return vo.getPromotionSessionId() + "_" + vo.getSkuId().toString();
                }).collect(Collectors.toList());
                redisTemplate.opsForList().leftPushAll(key, collect);
            }
        });
    }


    // 返回当前时间可以参与的秒杀商品信息
    @Override
    public List<SeckillSkuRedisTo> getCurrentSeckillSkus() {

        long now = new Date().getTime();
        Set<String> keys = redisTemplate.keys(SESSION_CACHE_PREFIX + "*");
        for (String key : keys) {
            System.out.println("key:" + key);
            String replace = key.replace(SESSION_CACHE_PREFIX, "");
            System.out.println("replace:" + replace);
            String[] split = replace.split("_");
            long startTime = Long.parseLong(split[0]);
            long endTime = Long.parseLong(split[1]);
            System.out.println("now:" + now);
            if (now > startTime && now < endTime) {
                // 获取这个场次需要的所有商品信息
                List<String> range = redisTemplate.opsForList().range(key, 0, -1);
                BoundHashOperations<String, String, Object> hashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
                List<Object> list = hashOps.multiGet(range);
                if (list != null || list.size() > 0) {
                    List<SeckillSkuRedisTo> collect = list.stream().map(item -> {
                        SeckillSkuRedisTo redisTo = JSON.parseObject((String) item, SeckillSkuRedisTo.class);
                        return redisTo;
                    }).collect(Collectors.toList());
                    return collect;
                }
            }
        }
        return null;
    }

    @Override
    public SeckillSkuRedisTo getSkuSecKillInfo(Long skuId) {
        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
        // 得到所有的key
        Set<String> keys = hashOps.keys();
        if (keys != null && keys.size() > 0) {
            String regx = "\\d_" + skuId;
            for (String key : keys) {
                boolean matches = Pattern.matches(regx, key);
                if (matches) {
                    // 判断是否在时间段内
                    String json = hashOps.get(key);
                    SeckillSkuRedisTo redisTo = JSON.parseObject(json, SeckillSkuRedisTo.class);
                    long startTime = redisTo.getStartTime().getTime();
                    long endTime = redisTo.getEndTime().getTime();
                    long nowTime = new Date().getTime();
                    if (nowTime > startTime && nowTime < endTime) {
                    } else {
                        redisTo.setUuid(null);
                    }
                    return redisTo;
                }
            }
        }
        return null;
    }

    @Override
    public void kill(String killId, String key, Integer num) {

        MemberVo memberVo = KillInteceptor.threadLocal.get();

        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
        String json = hashOps.get(killId);
        if(StringUtils.isNotEmpty(json)){
            SeckillSkuRedisTo redisTo = JSON.parseObject(json, SeckillSkuRedisTo.class);
            // 判断是否在规定的时间内
            long now = new Date().getTime();
            if(now >=redisTo.getStartTime().getTime() && now<redisTo.getEndTime().getTime()){
                // 判断随机码是否一致
                String skuId = redisTo.getPromotionSessionId()+"_"+redisTo.getSkuId();
                if(key.equals(redisTo.getUuid()) && killId.equals(skuId)){
                    // 判断购买数量是否超出范围
                    if(num<redisTo.getSeckillLimit().intValue()){
                        //判断是否已经购买过,setIfAbsent操作是原子性的，如果能够插入，则redis不含有此条数据（幂等性 ）
                        String redisKey = memberVo.getId()+"_"+skuId;
                        long ttl = redisTo.getEndTime().getTime() - redisTo.getStartTime().getTime();
                        Boolean ok = redisTemplate.opsForValue().setIfAbsent(redisKey, num.toString(),ttl, TimeUnit.MILLISECONDS);
                        if(ok){
                            //使用分布式信号量
                            RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + key);
                            //
                            boolean b = false;
                            try {
                                // 尝试获取锁
                                b = semaphore.tryAcquire(num,100, TimeUnit.MILLISECONDS);
                                if(b){
                                    // 给MQ发送消息
                                    String timeId = IdWorker.getTimeId();
                                    SecKillOrderTo killOrderTo = new SecKillOrderTo();
                                    killOrderTo.setOrderSn(timeId);
                                    killOrderTo.setMemberId(memberVo.getId());
                                    killOrderTo.setNum(num);
                                    killOrderTo.setPromotionSessionId(redisTo.getPromotionSessionId());
                                    killOrderTo.setSeckillPrice(redisTo.getSeckillPrice());
                                    rabbitTemplate.convertAndSend("order-event-exchange","order.seckill.order",killOrderTo);
                                }
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }
            }
        }
    }


}
